Generalize Snowflake's per-database discovery + filter-reason logging
(#28336) to a reusable helper used by every connector family —
Database, Dashboard, Pipeline, Messaging, Storage, MLModel. At end of
each source step, emit a single grep-friendly "FILTER VISIBILITY
REPORT" block listing the discovered count, every filtered name +
reason, and the kept count per entity type, so users can tell whether
a missing entity was removed by includes/excludes or never visible to
the ingestion role.

Storage profile is the diff only: discovered counts (int per type) +
the existing Status.filtered list (now carrying richer reasons). No
discovered-name or kept-name lists; both are derivable. No spec/JSON
schema change, no UI work.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

…efenses

Observability code in the ingestion hot path must not fail connectors.
Adds three layers of defense:

  1. Helper-internal try/except in log_discovered, log_filtered, and
     log_step_summary. Any failure is logged once at WARN and swallowed.
  2. Call-site try/except around log_step_summary in every base class
     close() (database, dashboard, pipeline, messaging, storage, mlmodel)
     so a summary failure can't mask the real cleanup work.
  3. Per-entity-type cap on Status.filtered (MAX_FILTERED_ENTRIES_PER_TYPE
     = 50_000). Past the cap, log_filtered only bumps a new
     Status.filtered_counts dict so the true count stays accurate without
     unbounded memory growth. The report annotates the truncation.

Test coverage doubled (12 -> 23): adds bounded-growth tests for the cap,
resilience tests for broken loggers / bad inputs / corrupted Status, and
an integration-style multi-database lifecycle test that verifies counts
are correct after the full streaming-log + summary cycle.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

- log_discovered: skip list() materialization when DEBUG logging is off.
  Use len() directly for Sized inputs (zero alloc) and a streaming
  sum() for generators. Cuts the per-discovery cost from O(n) memory
  to O(1) on large catalogs when --debug isn't active.

- _entity_type_from_reason: case-insensitive marker match so legacy
  reasons like "Database Filtered out" (lowercase 'out' from older
  BigQuery code paths) get bucketed into the right report section
  instead of being silently undercounted.

- dashboard_service.get_dashboard: null-safe via `or []` (matches the
  Optional[List[Any]] contract Pipeline / Messaging sources already
  use); materialize names once into a list so log_discovered can take
  the zero-allocation Sized path instead of re-listing a generator
  over the heavy dashboard objects.

- common_db_source.get_tables_name_and_type / dashboard_service: skip
  the O(n) shallow copy when query_*_names_and_types() / get_dashboards_list()
  already returned a list. Only materialize when a subclass returned
  a generator.

- Report: rename "Will be published to OpenMetadata" → "Passed filter
  patterns" and prepend a note explaining the count is filter-decision
  only (does not subtract source-side extraction errors or secondary
  filters like projectFilterPattern). Avoids confusion when reconciling
  report counts with the final ingested set.

Tests: 23 → 28 (added DEBUG-gated materialization, Sized len() path,
case-insensitive marker, and clarifying-note assertions).

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

- mark_databases_as_deleted: pass add_to_status=False to
  _get_filtered_database_names. Without this, every filtered database
  was being recorded twice (once during main ingestion via the
  per-connector get_database_names, once during the maintenance pass),
  inflating Status.filtered + filtered_counts and skewing the report.

- _get_filtered_schema_names: only materialize the raw schema names
  when add_to_status=True (where log_discovered needs the count up
  front). The mark-deleted maintenance paths iterate once with
  add_to_status=False — streaming saves O(n) memory on large schemas.
  Same fix applied to BigQuery's override.

- dashboard/pipeline/messaging get_X(): compute names once into
  (entity, name) tuples instead of calling the connector-specific
  get_*_name() twice per entity (once to build the discovery list,
  once inside the filter loop). Real savings when name extraction is
  non-trivial (e.g., Tableau workbook lookups).

- DB connector log_filtered calls: use the raw entity name (database_name,
  schema_name, table_name, view_name, new_database, project_id) as
  `name` and the FQN as `matched_against`. The previous convention of
  `name=<fqn>` produced confusing report output when useFqnForFiltering
  was False ("filtered name = FQN, matched against = raw name") and
  was inconsistent with non-SQL connectors (Dashboard/Pipeline/Topic/
  MLModel) which already stored raw names. Report column is now
  readable across all connector families.

Tests: existing 33 still pass (no test changes needed — the storage-key
change to raw names is verified by the existing dashboard/pipeline/topic
assertions which always asserted raw names).

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

Three categories of new basedpyright errors introduced by the
filter visibility refactor — all surfaced by the
`--baselinemode=discard` CI gate:

- log_step_summary: source_name was typed `str` but every call site
  passes `self.config.serviceName` which pyright infers as `str | None`.
  Widened to `str | None` and gracefully fall back to '<unknown source>'
  in the report header.

- filter_by_schema / filter_by_table / filter_by_database: I extracted
  the `filter_name = X if useFqn else Y` ternary into a variable across
  several connectors. The inline ternary was fine but the extracted
  variable became `str | None` to pyright (because one branch is an FQN
  typed as Optional). Widened these three helpers in filters.py to
  accept `Optional[str]` — they already delegate to `_filter()` which
  handles None per its own contract. Aligns the wrapper signatures
  with the actual delegate behavior; one-file fix instead of casting
  at 9 call sites.

- mlflow.get_mlmodels: the existing `cast(RegisteredModel, ...)` told
  pyright the search call returned a single model, so my `list(...)`
  wrap produced `list[tuple[str, Any]]` and broke `m.name` access.
  Fixed the cast to `cast(List[RegisteredModel], list(...))`. That
  surfaced a pre-existing latent bug: `model.latest_versions` is
  Optional but was iterated directly — guarded with `or []`.

Verification:
- `basedpyright -p pyproject.toml --baselinefile .basedpyright/baseline.json
  --baselinemode=discard` now reports zero new errors (the remaining
  errors are env-specific airflow/locust/pydoris import warnings
  already in the baseline; CI has those packages installed).
- `make py_format_check` clean.
- 79/79 helper + filter_pattern + common_db_source + snowflake tests
  still pass.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>